昨天的圖跑完一次就結束了,完全沒有記憶。你問它「我叫什麼名字」,它不會知道你上一句話有沒有自我介紹過,因為每次 graph.invoke(...) 都是全新的一次呼叫,State 用完就丟。
今天要解決這件事:讓使用者能跟教練進行多輪對話,而且就算把程式關掉、電腦重開機,之前聊過的內容還能接得回來。這對一個「陪你調整學習計畫」的教練來說是必要的,使用者不可能每次都要重新解釋一次自己的狀況。
昨天的圖,你每呼叫一次 graph.invoke(...),State 就是全新的一份,跑完就消失,跟上一次呼叫完全沒有關聯。
但真實情況是:同一個使用者今天問一句、明天再問一句,教練要知道這是「同一段對話的延續」。而且系統以後會有很多使用者,甲的對話不能混進乙的對話裡。
Thread 就是用來解決這件事:每一段獨立的對話給一個 Thread ID,LangGraph 用這個 ID 來區分「這是誰的對話、存到哪裡、下次要接回哪一份」。
想像成醫院的病歷夾:每個病人有自己的病歷號碼(Thread ID),每次回診,醫生報上病歷號碼,就能調出這個人完整的看診紀錄,不會跟別的病人搞混。
Checkpoint(檢查點)是 LangGraph 內建的自動存檔功能。只要在編譯圖的時候給它一個「checkpointer」,圖每跑完一步,就會自動把當下的 State 存起來,存的時候會標記是哪個 Thread ID 的。
下次你用同一個 Thread ID 呼叫這張圖,LangGraph 會先去讀回之前存的 State,接著才把新的輸入疊上去,而不是從空白開始。這整個過程你完全不用自己寫存檔、讀檔的邏輯,只要在編譯時指定 checkpointer 就好。
第一次呼叫(Thread ID: user-1)
↓
State 是空的 → 加入這次的訊息 → 跑完 → 自動存檔
第二次呼叫(Thread ID: user-1,同一個 ID)
↓
先讀回之前存的 State → 加入這次的新訊息 → 跑完 → 再存一次
checkpointer 可以存在記憶體裡(程式關掉就消失,適合開發時快速測試),也可以存進 SQLite 檔案(存在硬碟上,重開機也還在)。今天直接用 SQLite,一步到位。
Day 10 的 State 只有 user_input 和 ai_response 兩個欄位,一次只能裝一輪對話。多輪對話需要把每一輪都留下來,所以今天要把 State 換成一個「訊息清單」,每輪對話結束後用「新增」的方式加進清單,而不是覆蓋掉。
LangGraph 本身只內建「存在記憶體裡」的 checkpointer,要存進 SQLite 檔案需要另外裝一個套件:
cd backend
python -m pip install langgraph-checkpoint-sqlite
驗證:
python -c "from langgraph.checkpoint.sqlite import SqliteSaver; print('安裝成功')"
檔案位置: backend/graph.py
狀態: 修改檔案(整個改寫,取代 Day 10 的內容)
用途: 定義能累積多輪對話的 StateGraph,並加上 SQLite Checkpoint
依賴: langgraph, langgraph-checkpoint-sqlite, langchain-ollama
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class ChatState(TypedDict):
"""messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
messages: Annotated[list, add_messages]
Annotated[list, add_messages] 是關鍵:add_messages 是 LangGraph 提供的規則,告訴它「這個欄位每次更新時,是把新訊息接在後面,不是整份換掉」。沒有這個標記的話,每次 Node 回傳新訊息都會把舊的洗掉,記憶就沒了。
檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟2,繼續往下加)
用途: 定義呼叫 Ollama 的 Node,讓它一次看到完整對話歷史
依賴: langchain-ollama
from langchain_ollama import ChatOllama
_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)
def call_model(state: ChatState) -> dict:
"""把目前為止所有訊息一起丟給模型,讓它看得到完整對話歷史"""
response = _llm.invoke(state["messages"])
return {"messages": [response]}
跟 Day 10 的差別:這次丟給模型的是 state["messages"](一整串歷史訊息),不是單一一句話,模型才能「記得」前面聊過什麼。
檔案位置: backend/graph.py
狀態: 修改檔案(接續步驟3,繼續往下加)
用途: 加上 SQLite Checkpoint,組建並編譯出最終的 graph
依賴: langgraph, langgraph-checkpoint-sqlite
import sqlite3
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
# check_same_thread=False:允許這個連線被不同的呼叫共用,Checkpoint 內部需要這個設定
_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", call_model)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)
# 編譯時把 checkpointer 傳進去,圖就會自動存檔
graph = graph_builder.compile(checkpointer=_checkpointer)
checkpoints.sqlite 是一個新的檔案,跟 Day 5 的 app.db 是不同的資料庫,一個專門存教練的使用者資料,一個專門存 LangGraph 的對話記憶,兩者互不相干。
完整的 graph.py 到這裡長這樣:
import sqlite3
from typing import Annotated
from typing_extensions import TypedDict
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
class ChatState(TypedDict):
"""messages 是一個清單,每輪對話會被「加進去」而不是覆蓋掉"""
messages: Annotated[list, add_messages]
_llm = ChatOllama(model="llama3.1:8b", temperature=0.3)
def call_model(state: ChatState) -> dict:
"""把目前為止所有訊息一起丟給模型,讓它看得到完整對話歷史"""
response = _llm.invoke(state["messages"])
return {"messages": [response]}
_conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
_checkpointer = SqliteSaver(_conn)
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", call_model)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)
graph = graph_builder.compile(checkpointer=_checkpointer)
這支腳本每次執行都是「獨立的一個 Python 程式」,模擬使用者關掉終端機、下次再打開的情境。用固定的 Thread ID,證明對話歷史真的有被存下來。
檔案位置: backend/chat_once.py
狀態: 新增檔案
用途: 每次執行送一句話進去,印出目前為止的完整對話歷史
依賴: graph
import sys
from graph import graph
def main() -> None:
"""從命令列拿一句話,用固定的 Thread ID 送進圖裡"""
if len(sys.argv) < 2:
print('用法:python chat_once.py "你要說的話"')
return
user_message = sys.argv[1]
config = {"configurable": {"thread_id": "demo-user"}}
result = graph.invoke(
{"messages": [{"role": "user", "content": user_message}]}, config
)
print("目前完整對話:")
for msg in result["messages"]:
print(f" [{msg.type}] {msg.content}")
if __name__ == "__main__":
main()
先跑第一句:
python chat_once.py "我叫Alice,正在準備AWS Solutions Architect認證"
應該看到一問一答:
目前完整對話:
[human] 我叫Alice,正在準備AWS Solutions Architect認證
[ai] 你好Alice!準備AWS SA認證是個很棒的目標...
重點在這裡:完全關掉這個終端機視窗,開一個全新的終端機視窗,進入同一個 backend/ 目錄,跑第二句話:
python chat_once.py "我剛剛說我叫什麼名字?"
應該看到完整的四則訊息,而且模型的回答會正確答出「Alice」:
目前完整對話:
[human] 我叫Alice,正在準備AWS Solutions Architect認證
[ai] 你好Alice!準備AWS SA認證是個很棒的目標...
[human] 我剛剛說我叫什麼名字?
[ai] 你說你叫Alice。
這證明了兩件事:對話歷史真的存進了 checkpoints.sqlite,不是只存在記憶體裡;就算是完全獨立的兩次程式執行(等同於關掉重開),只要 Thread ID 一樣,歷史就接得回來。
先確認兩次執行用的是同一個 thread_id(範例裡固定寫死 "demo-user"),還有兩次都在同一個 backend/ 目錄下執行,確保用的是同一個 checkpoints.sqlite 檔案。
sqlite3.OperationalError: database is locked通常是同時有兩個程式在寫同一個 checkpoints.sqlite。確認前一個 python chat_once.py 已經執行完畢、終端機已經回到提示字元,再執行下一次。
checkpoints.sqlite 檔案在哪裡?在 backend/ 目錄下,第一次執行 chat_once.py 時會自動建立,可以用檔案總管或 ls 確認它存在。
ModuleNotFoundError: No module named 'langgraph.checkpoint.sqlite'確認用 python -m pip install langgraph-checkpoint-sqlite 這個獨立套件,光裝 langgraph 本身是不夠的,SQLite 的 checkpoint 功能是分開打包的。
Annotated[list, add_messages] 這個寫法看不懂把它想成一個特殊標記:告訴 LangGraph「這個欄位是清單,而且更新方式是『加在後面』」。不加這個標記的話,messages 欄位每次會被整個覆蓋掉,對話就會斷片,只記得最後一輪。
今天讓教練真正有了記憶。改寫了 State 讓它能累積多輪對話,加上 SQLite Checkpoint 讓對話存進硬碟,用兩次完全獨立的程式執行證明了記憶真的能跨越「關掉重開」這件事。
系統現在是這樣的:
Day 1 ✓ 產品定義完成
Day 2 ✓ 開發環境準備
Day 3 ✓ 專案架構設計
Day 4 ✓ 資料庫設計
Day 5 ✓ SQLite 資料庫建置
Day 6 ✓ FastAPI 基礎
Day 7 ✓ 使用者檔案 API
Day 8 ✓ 理解 LLM Agent 的本質
Day 9 ✓ 連接 Ollama 本機模型
Day 10 ✓ LangGraph 最小範例
Day 11 ✓ Thread 與 State 管理(今天)
Day 12 ⬜ 簡化的意圖路由
有了記憶之後,教練才有辦法真的「認得」使用者。明天(Day 12)要教它聽懂使用者在說什麼:這句話是要建立新計畫、回報進度,還是單純問問題,讓圖依照不同意圖走不同路線,這也是圖上第一次出現「分岔路」。